package kafka.loganalysis;

import com.alibaba.fastjson.JSONObject;
import kafka.KafkaProperties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author: pizhihui
 * @datae: 2017-11-09
 */
public class LogGenerator {
    private static final Logger logger = LoggerFactory.getLogger(LogGenerator.class);
    public static final String LOG_TOPIC = "log";


    private static void logGeneratorRunWithOldApi() {
        ProducerConfig producerConfig = new ProducerConfig(setKafkaProperties());
        Producer<Object, String> producer = new Producer<>(producerConfig);

        while (true) {
            KeyedMessage<Object, String> message = new KeyedMessage<>(LOG_TOPIC, generateLogMsg());
            logger.error("send message: {}", ToStringBuilder.reflectionToString(message, ToStringStyle.SHORT_PREFIX_STYLE));
            producer.send(message);
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                logger.error("error: {}", e);
            }

        }
    }

    private static Properties setKafkaProperties() {
        Properties props = new Properties();

        props.put("metadata.broker.list", KafkaProperties.BROKER_CONNECT);
        //props.put("partitioner.class", "com.cuicui.kafkademon.MyPartitioner");
        props.put("request.required.acks", "1");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        return props;
    }

    private static String generateLogMsg() {

        Map<Object, Object> map = new HashMap<>();
        map.put("uid", getUserId());
        map.put("event_time", System.currentTimeMillis());
        map.put("os_type", getOSType());
        map.put("click_count", getClick());

        return JSONObject.toJSONString(map);
    }

    private static String getUserId() {
        return RandomStringUtils.randomNumeric(20).toUpperCase();
    }

    private static int getClick() {
        Random random = new Random();
        return random.nextInt(10);
    }

    private static String getOSType() {
        String[] arr = new String[]{"IOS", "Android", "Symban", "else"};
        return arr[RandomUtils.nextInt(0, 4)];
    }



    public static void main(String[] args) throws InterruptedException {
        if (!KafkaTopicUtils.existsTpic(LOG_TOPIC)) {
            KafkaTopicUtils.createTopic(LOG_TOPIC);
        }
        logGeneratorRunWithOldApi();
    }

}
